package com.atguigu.pro2

import java.net.URL
import java.util

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
 * @description: xxx
 * @time: 2021/4/6 10:29
 * @author: baojinlong
 * */

// 输入的登录事件样例类
case class LoginEvent(userId: Long, ip: String, eventType: String, timestamp: Long)

// 输出报警信息样例类
case class LoginFailWarning(userId: Long, firstFailTime: Long, lastFailTime: Long, warningMsg: String)

object LoginFail {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(1)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/LoginLog.csv")
    //val inputStream: DataStream[String] = environment.readTextFile(resource.getPath)
    val inputStream: DataStream[String] = environment.readTextFile("src/main/resources/LoginLog.csv")
    // 转换成样例类行,并提起时间戳和watermark

    val loginEventStream: DataStream[LoginEvent] = inputStream
      .map(data => {
        val arr: Array[String] = data.split(",")
        LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      // 3秒延时
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(t: LoginEvent): Long = t.timestamp * 1000
      })

    // 进行判断和检测,如果2秒之内连续登录失败,输出报警信息
    val loginFailWarningStream: DataStream[LoginFailWarning] = loginEventStream
      .keyBy(_.userId)
      .process(new LoginFailWarningResult(2))

    loginFailWarningStream.print("loginFailWarningStream")
    environment.execute("loginFailWarningStreamTest")
  }
}

// 连续登录failTimes就报警
class LoginFailWarningResult(failTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, LoginFailWarning] {
  // 定义状态,保存当前所有的登录失败事件,保存定时器的时间戳
  lazy val loginFailListState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("loginFailListState", classOf[LoginEvent]))
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timerTsState", classOf[Long]))

  override def processElement(loginEvent: LoginEvent, context: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#Context, collector: Collector[LoginFailWarning]): Unit = {
    // 判断当前登录事件是成功还是失败
    if ("fail".equals(loginEvent.eventType)) {
      // 只要是登录失败则添加到集合中
      loginFailListState.add(loginEvent)
      if (timerTsState.value == 0) {
        // 如果没有定时器,那么注册一个2秒后的定时器
        val ts: Long = loginEvent.timestamp * 1000 + 2000
        context.timerService.registerEventTimeTimer(ts)
        timerTsState.update(ts)
      }
    } else {
      // 如果是成功那么直接定时器和清空状态
      context.timerService.deleteEventTimeTimer(timerTsState.value)
      loginFailListState.clear()
      timerTsState.clear()
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#OnTimerContext, out: Collector[LoginFailWarning]): Unit = {
    // 判断登录失败的次数,简化写法
    //val list = List(loginFailListState.get.iterator)
    val allLoginFailList: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]
    val iterator: util.Iterator[LoginEvent] = loginFailListState.get.iterator
    while (iterator.hasNext) {
      allLoginFailList += iterator.next
    }
    // 判断登录失败事件个数,如果超过上限则报警
    if (allLoginFailList.length > failTimes) {
      val headValue: LoginEvent = allLoginFailList.head
      out.collect(LoginFailWarning(headValue.userId, headValue.timestamp, allLoginFailList.last.timestamp, "login fail in 2s for " + allLoginFailList.length + " times"))
    }
    // 状态清空
    loginFailListState.clear()
    timerTsState.clear()
  }
}
